package com.xiangxiao.rpan.data.listener.consumer;

import com.alibaba.fastjson.JSONObject;
import com.xiangxiao.rpan.data.enums.MessageTypeEnum;
import com.xiangxiao.rpan.data.factory.ApplicationContextFactory;
import com.xiangxiao.rpan.data.model.BaseAlarm;
import com.xiangxiao.rpan.data.producer.AlarmProduce;
import com.xiangxiao.rpan.data.threadpool.AsyncTaskWithThreadPool;
import org.reflections.Reflections;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.File;
import java.lang.reflect.Constructor;
import java.lang.reflect.Modifier;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @auther xiangxiao
 * @email 573768011@qq.com
 * @data 2024/7/22 15:56
 * 1.维护了一个 Map 来存储队列和位点的关系。每个消费者线程都有一个对应的位点来记录消费的进度。
 * 在 commitOffset 方法中，根据 key 提交位点的偏移值;
 * 2.消费线程在每次成功消费一条数据后，更新位点，并判断是否满足提交位点的条件,这里模拟每消费10条数据提交一次位点
 */
public class RpanDataConsumer {
  private Map<String, BlockingQueue<BaseAlarm>> queues;
  private Map<String, Integer> offsets;
  private Map<String, Runnable> threads;

  private AsyncTaskWithThreadPool asyncTaskExecutor;

  private static RpanDataConsumer instance;

  private static List<String> topics = MessageTypeEnum.getAllMQqttTopics();

  private RpanDataConsumer(int threadNum) {
    queues = new HashMap<>();
    offsets = new HashMap<>();
    threads = new HashMap<>();
    asyncTaskExecutor = new AsyncTaskWithThreadPool();

    for (int i = 0; i < threadNum; i++) {
      BlockingQueue<BaseAlarm> queue = new LinkedBlockingQueue<>();
      String key = Integer.toString(i);
      queues.put(key, queue);
      offsets.put(key, 0);

      // 创建消费线程, 放入map
      Runnable consumerRunnable = new CustomConsumerThread(queue, key);
      threads.put(key, consumerRunnable);
    }
    ;
  }

  public void consumeData(Date alarmTime, BaseAlarm data) {
    Integer key = 0;
    Integer seconds = alarmTime.getSeconds();
    if (seconds < 15) {
      key = 0;
    } else if (seconds >= 15 && seconds < 30) {
      key = 1;
    } else if (seconds >= 30 && seconds < 45) {
      key = 2;
    } else if (seconds >= 45) {
      key = 3;
    }
    BlockingQueue<BaseAlarm> queue = queues.get(Integer.toString(key));
    Runnable currentConsumer = threads.get(Integer.toString(key));
    if (queue != null) {
      try {
        // 将数据放入对应的队列
        queue.put(data);
        asyncTaskExecutor.asyncServiceExecutor().submit(currentConsumer);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
    }
  }

  public void commitOffset(String key, int offset) {
    offsets.put(key, offset);
    System.out.println("Committed offset for key " + key + ": " + offset);
  }

  public static synchronized RpanDataConsumer getInstance() {
    if (instance == null) {
      instance = new RpanDataConsumer(topics.size());
    }
    return instance;
  }

  /**
   * 消费消息线程类
   */
  class CustomConsumerThread implements Runnable {
    private final BlockingQueue<BaseAlarm> queue;
    private final String key;
    private int offset;
    private List<AlarmProduce> alarmProduceList;

    public CustomConsumerThread(BlockingQueue<BaseAlarm> queue, String key) {
      this.queue = queue;
      this.key = key;
      this.offset = 0;
    }

    public List<AlarmProduce> getAllImplementations() {
      List<AlarmProduce> intances = new ArrayList<>();
      Reflections relfs = new Reflections("com.xiangxiao.rpan.data.producer");
      try {
        Set<Class<? extends AlarmProduce>> subTypesOf2 = relfs.getSubTypesOf(AlarmProduce.class);
        for (Class<? extends AlarmProduce> aClass : subTypesOf2) {
          System.out.println(aClass.getName());
          try {
            AlarmProduce instance = aClass.newInstance();
            intances.add(instance);
          } catch (InstantiationException e) {
            e.printStackTrace();
          } catch (IllegalAccessException e) {
            e.printStackTrace();
          }
        }
      } catch (Exception e) {
        e.printStackTrace();
      } finally {
        return intances;
      }
    }

    @Override
    public void run() {
      // 消费队列中的数据
      System.out.println(Thread.currentThread().getName() + ", non-interrupted status is " + !Thread.currentThread().isInterrupted());
      List<AlarmProduce> alarmProduceList = getAllImplementations();
      while (!Thread.currentThread().isInterrupted()) {
        try {
          BaseAlarm data = queue.take();
          alarmProduceList.stream()
              .filter(alarmProduce -> alarmProduce.getMessageTypeEnum().getCode().equals(data.getAlarmType()))
              .forEach(alarmProduce -> {
                alarmProduce.alarm(data);
              });
          offset++;

          System.out.println("Commit offset: " + offset);
          // 模拟提交位点
          if (offset % 4 == 0) {
            RpanDataConsumer.getInstance().commitOffset(key, offset);
          }
        } catch (InterruptedException e) {
          e.printStackTrace();
          Thread.currentThread().interrupt();
        }
      }
    }
  }
}

